package com.chukun.flink.stream.action.click;

import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.HashSet;
import java.util.Iterator;
import java.util.Properties;
import java.util.Set;

/**
 * @author chukun
 * @version 1.0.0
 * @description pv 与 uv 统计
 * @createTime 2022年05月21日 21:54:00
 */
public class PVAndUVAnalyse {

    /**
     * 迟到的用户事件
     */
    private static final OutputTag<UserBehaviorEvent> USER_LATENESS_EVENT = new OutputTag<UserBehaviorEvent>("user_lateness_event") {};

    public static void main(String[] args) throws Exception {

        /**
         * kafka配置信息
         */
        Properties kafkaProps = new Properties();
        kafkaProps.setProperty("bootstrap.servers", Constants.KAFKA_BROKERS);

        /**
         * kafka消费者信息
         */
        FlinkKafkaConsumer010<UserBehaviorEvent> kafkaConsumer = new FlinkKafkaConsumer010<>(Constants.KAFKA_TOPIC, new UserBehaviorEventSchema(), kafkaProps);
        // 从最新消息处理
        kafkaConsumer.setStartFromLatest();
        // 不设置checkpoint
        kafkaConsumer.setCommitOffsetsOnCheckpoints(false);

        // 获取flink的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置基于kafka的事件时间处理,新版api不需要设置
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // 设置kafka事件时间生成策略，watermarker商城策略
        kafkaConsumer.assignTimestampsAndWatermarks(
                WatermarkStrategy.forGenerator((ctx) -> new PeriodicWatermarkGenerator())
                        .withTimestampAssigner((ctx) -> new TimeStampExtractor()));

        // 添加kafka数据流
        DataStream<UserBehaviorEvent> dataStreamByEventTime = env.addSource(kafkaConsumer);

        // 基于kafka数据流处理
        SingleOutputStreamOperator<Tuple4<Long, Long, Long, Integer>> processStream = dataStreamByEventTime
                // 设置窗口为滑动窗口
                .windowAll(SlidingEventTimeWindows.of(Time.seconds(5), Time.seconds(1)))
                // 容忍时间延时时间
                .allowedLateness(Time.seconds(5))
                .sideOutputLateData(USER_LATENESS_EVENT)
                .process(new ProcessAllWindowFunction<UserBehaviorEvent, Tuple4<Long, Long, Long, Integer>, TimeWindow>() {
                    @Override
                    public void process(ProcessAllWindowFunction<UserBehaviorEvent, Tuple4<Long, Long, Long, Integer>, TimeWindow>.Context context, Iterable<UserBehaviorEvent> iterable, Collector<Tuple4<Long, Long, Long, Integer>> collector) throws Exception {
                        long pv = 0L;
                        Set<Integer> userIds = new HashSet<>();
                        Iterator<UserBehaviorEvent> it = iterable.iterator();
                        while (it.hasNext()) {
                            UserBehaviorEvent event = it.next();
                            pv++;
                            userIds.add(event.getUserId());
                        }
                        TimeWindow currentWindow = context.window();
                        collector.collect(Tuple4.of(currentWindow.getStart(), currentWindow.getEnd(), pv, userIds.size()));
                    }
                });

        // 获取迟到的用户事件流
        DataStream<UserBehaviorEvent> latenessStream = processStream.getSideOutput(USER_LATENESS_EVENT);

        // 打印迟到的数据流
        latenessStream.print("user.event.lateness").setParallelism(1);

        // 打印正常的数据流
        processStream.print("user.event.normal").setParallelism(1);

        env.execute("PVAndUVAnalyse");

    }
}
